package com.demo.webflux.dubbo;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.dubbo.demo.Student;
import com.demo.webflux.Constant;
import com.demo.webflux.PropertiesInnerUtil;

import dev.miku.r2dbc.mysql.MySqlConnectionConfiguration;
import dev.miku.r2dbc.mysql.MySqlConnectionFactory;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import reactor.core.publisher.Mono;

@RestController
public class HelloController {

	private static Logger logger = LoggerFactory.getLogger(HelloController.class);
	private static String userName = PropertiesInnerUtil.getString(Constant.DATASOURCE_LOCATION, "datasource.username");
	private static String password = PropertiesInnerUtil.getString(Constant.DATASOURCE_LOCATION, "datasource.password");
	private Mono<Connection> connectionMono;
	@Autowired
	private DubboClientService dubboClientService;

	public HelloController() {
		MySqlConnectionConfiguration.Builder builder = MySqlConnectionConfiguration.builder()
	            .host("10.16.9.34")
	            .port(3306)
	            .connectTimeout(Duration.ofSeconds(3))
	            .username(userName)
	            .password(password)
	            .database("sms_crm_db");

		MySqlConnectionConfiguration configuration = builder.build();
		ConnectionFactory connectionFactory = MySqlConnectionFactory.from(configuration);
		connectionMono = Mono.from(connectionFactory.create());
	}

	@GetMapping("/dubbo")
    public Mono<Student> dubbo() {
    	logger.info("main");
    	Mono<Student> result = connectionMono.flatMapMany(connection -> {
    		logger.info("select");
			return connection.createStatement("SELECT cid, logic_code, message_send_code, message_content FROM message_queue_receiver limit 1").execute();
		})
    	.flatMap(rs -> {
    		logger.info("flatMap");
    		return rs.map((row, metadata) -> row.get(1, Object.class));
    	})
    	.collectList()
    	.flatMap(list -> {
    		Object data = list.get(0);
    		Student st = new Student();
    		st.setName("zsl" + data);
    		st.setAge(10);

    		// 这种方式本质上还是阻塞，阻塞在dubboClientService的Future.get()上
    		return Mono.create(sink -> {
	           CompletableFuture<Student> future = CompletableFuture.supplyAsync(() -> {
	        	   try {
	        		   return dubboClientService.saveUser(st);
	        	   } catch (InterruptedException | ExecutionException e) {
						e.printStackTrace();
	        	   }
	        	   return null;
	           });
	           future.whenComplete((v, t) -> {
	               if(t != null) {
	                   sink.error(t);
	               } else {
	                   sink.success(v);
	               }
	           });
	        });
    	});
    	logger.info("main end");
    	return result;
    }

	@GetMapping("/dubboAsync")
    public Mono<Student> dubboAsync() {
    	logger.info("main");
    	Mono<Student> result = connectionMono.flatMapMany(connection -> {
    		logger.info("select");
			return connection.createStatement("SELECT cid, logic_code, message_send_code, message_content FROM message_queue_receiver limit 1").execute();
		})
    	.flatMap(rs -> {
    		logger.info("flatMap");
    		return rs.map((row, metadata) -> row.get(1, Object.class));
    	})
    	.collectList()
    	.flatMap(list -> {
    		Object data = list.get(0);
    		Student st = new Student();
    		st.setName("zsl" + data);
    		st.setAge(10);

    		return dubboClientService.saveUserAsync(st);
    	});
    	logger.info("main end");
    	return result;
    }

}